import uuid
import json
from itertools import islice
from typing import List, Union
import asyncio
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from fastapi.params import Depends, Body
from redis import Redis
from sqlalchemy import or_,and_
from sqlalchemy.orm import Session
from starlette import status
from Model.default import  get_db
from Model.mysocketio.info import Group_Room, Double_Room, Double_Message, Request_Room
from Model.mysocketio.response.Double_Room import Double_Room_Public
from Model.mysocketio.response.Group_Room import Group_Room_Public
from Model.notice_message.main import no_response
from Model.user.User import user_expand, user
from Model.user.response.main import user_info_public
from MyRedis.main import get_redis_client, parse_redis_data, get_last_message_id
from api.mysocketio.create_Room import create_room
from api.mysocketio.handler import  MyPersistentRoomManager
from api.mysocketio.onlinehandler import ONLINE_SENDER
from api.mysocketio.update_room_info import ADD_ROOM_USER_MAPPER
from api.user import get_token_from_cookie, get_current_user
from celerywork.Chat_ import deal_create_request_room, deal_response_request_room
from config.chat import GET_MESSAGE_LENGTH, GET_USER_SEARCH_LENGTH, GET_ROOM_REQUEST_LENGTH, \
    GET_MESSAGE_NORESPONSE_LENGTH
from config.scoket_chat import GROUP_USER_INFO_LENGTH, SEARCH_USER_LENGTH

# 创建 WebSocket 路由
mysocket_chat = APIRouter(prefix='/chat_socket', tags=['chat_sockets'],)
@mysocket_chat.websocket("/ws/{user_id}/{jwt}")
async def websocket_endpoint(websocket: WebSocket, jwt:str,user_id:int,db:Session=Depends(get_db)):
    # 处理连接事务
    websocket.user_id= user_id
    await  get_current_user(jwt[1:len(jwt)-1])
    # 前端问题有,直接解析出有""
    await  MyPersistentRoomManager.connect(websocket)
    try:
        # 先存储用户的ip和端口用于减少操作,用于推送服务
        db.query(user_expand).filter_by(id=user_id).first().socket=json.dumps({
            "ip":websocket.client.host,
            "port":websocket.client.port
        })
        db.commit()
        db.close()
        while True:
            try:
                data = await websocket.receive_json()
                # 前端发送的信息应该是:
                #   {
                #       type:
                #       data:{
                #             字段:
                #             字段:
                #             }
                #   }
                #

                # print(data.get("type"))
                data["data"]['user_id']= user_id
                data["data"]["type"]=data.get("type")
                # 如果是客户端主动发的心跳响应，直接跳过处理
                if data.get("type") == "ping":
                    # 让心跳机制能够执行
                    await websocket.send_json({"type":'ping',"data":'pong'})
                    continue
                # 正常业务逻辑
                await MyPersistentRoomManager.Handler.handle(websocket, data['data'])
            except asyncio.TimeoutError:
                # 心跳检测
                if not await MyPersistentRoomManager.check_connection(websocket):
                    break
                continue
            except RuntimeError as e:
                if "WebSocket connection is closed" in str(e):
                    break
                raise
    except WebSocketDisconnect:
        # 先断开x
        MyPersistentRoomManager.disconnect(websocket=websocket)
        print(f"用户 {user_id} 断开连接")
    except Exception as e:
        print(f"WebSocket 错误: {e}")
    finally:
        print(f"用户 {user_id} 连接已关闭")


async def sender_message_online_By_Type(user_id,type,**kwargs):
    try:
        # 如何推送由具体情况处理
        ws=MyPersistentRoomManager.active_connections.get(user_id,None)
        if ws:
            await ONLINE_SENDER.get(type)(ws,**kwargs)
    except Exception as e:
        print(e)
        pass

# 用于直接的专门用于用户的渲然,这个只用于添加时
async def sender_message_user_online_room_request(user_id,room_id,room_type):
    #由于需要db所以单独拿出来,用于用户同意申请后,通知用户。
    try:
        ws = MyPersistentRoomManager.active_connections.get(user_id, None)
        if ws and room_type=='double':
            db=get_db()
            room_info=db.query(Double_Room).filter_by(room_id=room_id).first()
            data={
                    "friend_info":user_info_public.from_orm(
                        db.query(user).filter(user.id!=user_id,user.id.in_([room_info.one,room_info.another])).first()).to_dict(),
                    "room_info":room_info.to_dict()
            }
            await ws.send_json({"type":'render_user_group',
                               "data":{
                                    'room_info':data,
                                    'room_id':room_id,
                                    'room_type':room_type
                                 }})
        elif ws and room_type=='group':
            db = get_db()
            data = {
                "room_info": db.query(Group_Room).filter_by(room_id=room_id).first().to_dict()
            }
            await ws.send_json({"type": 'render_user_group',
                                "data": {
                                    'room_info': data,
                                    'room_id': room_id,
                                    'room_type': room_type
                                }})
    except Exception as e:
        print(e)
        pass
    finally:
        db.close()




@mysocket_chat.post("/ws/set/room/create")
async  def deal_add_userinfo_to_room(created_user:int=0,user_id:List=Body(...,embed=True),type:str=Body(...,embed=True),):
    """
    根据类型创建不同的,聊天标记房间，实际只有群聊使用这个api
    """
    try:
        result=create_room(user_id,type,created_user)
        if not result:
            return status.HTTP_404_NOT_FOUND
        for i in user_id:
            asyncio.create_task(sender_message_online_By_Type(i, "render_groups_list"))
        return status.HTTP_200_OK

    except Exception as e:
        print(e)
        pass




@mysocket_chat.post("/ws/find/user")
def deal_find_user(key:Union[str,int],limit:int=1,db:Session=Depends(get_db)):
    """
    用于搜索用户,待定
    :param key:
    :param limit:
    :param db:
    :return:
    """
    try:
        DATA=[]
        if type(key)==int:
            pre_Data=db.query(user).filter_by(id=key).offset((limit-1)*SEARCH_USER_LENGTH).limit(SEARCH_USER_LENGTH).all()
        else:
            pre_Data=db.query(user).filter(user.username.ilike(f"%{key}%")).limit(limit * SEARCH_USER_LENGTH).all()
        for i in pre_Data:
            DATA.append(user_info_public.from_orm(i))
        return DATA
    except Exception as e:
        print(e)
        return  status.HTTP_400_BAD_REQUEST
    finally:
        db.close()


@mysocket_chat.post("/ws/user/friend/add")
async def deal_add_friends(user_id:int,friend_id:int,db:Session=Depends(get_db)):
    """
    用于搜索后的添加好友功能,先申请,在处理后续逻辑
    :param user_id:
    :param friend_id:
    :return:
    """
    try:
        pre_judge_one=db.query(Double_Room).filter(or_(
                    and_(Double_Room.one == user_id, Double_Room.another == friend_id,),
                    and_(Double_Room.one == friend_id, Double_Room.another == user_id)
                )
            ).limit(1).first()
        if pre_judge_one :
            # 存在好友关系
            return status.HTTP_400_BAD_REQUEST
        #一周过期
        pre_judge_two= db.query(Request_Room).filter(
                or_(
                    and_(Request_Room.one == user_id, Request_Room.another == friend_id),
                    and_(Request_Room.one == friend_id, Request_Room.another == user_id)
                )
            ).limit(1).first()
        # 这里这样做是由于python解释器解释 if 下的 if 为and
        if not pre_judge_two:
            pass
        elif pre_judge_two.status == 0:

            # 有记录且 status=0，禁止重复申请
            return status.HTTP_400_BAD_REQUEST
        else:
            db.delete(pre_judge_two)
            db.commit()
            #先删除上次记录，由于celery无法传输db,他需要json化暂且删除处理

        judge =deal_create_request_room([user_id,friend_id],'double',user_id)
        #任务id
        if judge:
            friend=db.query(user).filter_by(id=friend_id).first()
            message = no_response(id=str(uuid.uuid4()), user_id=user_id,
                                  title="好友申请通知",
                                  content=f'您向用户 :{friend.username or friend.id} 的好友申请,发送成功!',
                                  sender_info="系统通知"
                                  )
            db.add(message)
            db.commit()
            return  status.HTTP_200_OK
        else:
            return status.HTTP_400_BAD_REQUEST

    except Exception as e:
        print(e)
        db.rollback()
        return status.HTTP_404_NOT_FOUND
    finally:
           db.close()
           asyncio.create_task (sender_message_online_By_Type(friend_id,"request_message",))
           asyncio.create_task (sender_message_online_By_Type(user_id, "render_notice"))


@mysocket_chat.post("/ws/user/friends/delete")
async def deal_delete_user_friend(user_id: int, friend_id: int, db: Session = Depends(get_db)):
       """
       用于好友删除
       :param user_id:
       :param friend_id:
       :param db:
       :return:
       """
       try:
            users = db.query(user_expand).filter(user_expand.id.in_([user_id, friend_id])).all()
            one, another = (users[0], users[1]) if users[0].id == user_id else (users[1], users[0])
            room = db.query(Double_Room).filter(
                or_(
                    and_(Double_Room.one == user_id, Double_Room.another == friend_id,),
                    and_(Double_Room.one == friend_id, Double_Room.another == user_id)
                )
            ).first()
            if room:
                for USER in (one, another):

                    userinfo = db.query(user).filter_by(id=USER.user_id).first()
                    userinfo.friends_count -= 1

                    streaminfo = json.loads(USER.last_stream or "{}")
                    streaminfo.pop(room.room_id, None)
                    USER.last_stream = json.dumps(streaminfo)

                    roominfo = json.loads(USER.room_info or "{}")
                    roominfo.pop(room.room_id, None)
                    USER.room_info = json.dumps(roominfo)

                    friendinfo = json.loads(USER.friends_info or "{}")
                    friend_id = str(another.user_id if userinfo.id == one.user_id else one.user_id)
                    friendinfo.pop(friend_id, None)
                    USER.friends_info = json.dumps(friendinfo)
                db.delete(room)
                db.commit()
            return status.HTTP_200_OK
       except Exception as e:
            print(e)
            db.rollback()
            return status.HTTP_400_BAD_REQUEST
       finally:
            db.close()
            await  sender_message_online_By_Type(user_id, "render_notice", )
            # 此处应该预留向权限群聊发送申请处理




@mysocket_chat.post("/ws/user/group/add")
async def deal_add_groups(user_id:int,room_id:str,db:Session=Depends(get_db)):
   """
   用于主动:用户的群申请
   :param user_id:
   :param room_id:
   :param db:
   :return:
   """
   try:
        #一周过期
        prejudge1=db.query(Group_Room).filter_by(room_id=room_id).first()
        if json.loads(prejudge1.user_info or json.dumps({})).get(user_id,None):
            # 已经在群聊
            return status.HTTP_400_BAD_REQUEST
        prejudge2=db.query(Request_Room).filter_by(room_id=room_id).first()
        if prejudge2:
            pass
        elif prejudge2.status ==0:
            return status.HTTP_400_BAD_REQUEST
        else:
            db.delete(prejudge2)
            db.commit()

        judge = deal_create_request_room(user_id,'group',user_id,room_id)
        if not judge:
            return status.HTTP_404_NOT_FOUND
        message=no_response(id=str(uuid.uuid4()),user_id=user_id,
                            title="群申请通知",
                            content=f'您加入群聊 :{prejudge1.group_name or prejudge1.room_id }的申请已经发送!',
                            sender_info="系统通知"
                            )
        db.add(message)
        return status.HTTP_200_OK
   except Exception as e:
        print(e)
        return status.HTTP_404_NOT_FOUND
   finally:
        db.close()
        await  sender_message_online_By_Type(user_id, "render_notice", )

@mysocket_chat.post("/ws/user/group/quit")
def deal_quit_groups(user_id:int, room_id:str, db:Session=Depends(get_db), ):
    """
    用于用户退出群聊
    :param user_id:
    :param room_id:
    :param db:
    :return:
    """
    try:
        RoomInfo=db.query(Group_Room).filter_by(room_id=room_id).first()
        UserExpand=db.query(user_expand).filter_by(user_id=user_id).first()
        User=db.query(user).filter_by(id=user_id).first()


        room_to_user=json.loads(RoomInfo.user_info or json.dumps({}))
        room_to_user.pop(str(user_id), None)
        RoomInfo.user_info = json.dumps(room_to_user)

        user_to_room=json.loads(UserExpand.room_info or json.dumps({}))
        user_to_room.pop(RoomInfo.room_id, None)
        UserExpand.room_info=json.dumps(user_to_room)

        room_stream=json.loads(UserExpand.last_stream or json.dumps({}))
        room_stream.pop(RoomInfo.room_id, None)

        UserExpand.last_stream=json.dumps(room_stream)
        User.groups_count = max(User.groups_count, 0)

        db.commit()

        return status.HTTP_200_OK
    except Exception as e:
        db.rollback()
        print(e)
        return status.HTTP_400_BAD_REQUEST
    finally:
        db.close()

@mysocket_chat.get("/ws/get/room/request")
def deal_request_to_room(user_id:int,limit:int=1,db:Session=Depends(get_db)):
    """
    用于获取用户的申请列表
    :param user_id:
    :param room_id:
    :param db:
    :return:
    """
    try:
        # 目前仅仅做简单的获取处理
        # 先查询总记录数（不包含分页）
        # 暂时只返回需要处理的
        total_count = db.query(Request_Room).filter(
            and_(
                Request_Room.user_id != user_id,
                or_(Request_Room.one==user_id,Request_Room.another==user_id),
                Request_Room.status != 2,Request_Room.status!=1)
        ).count()
        Data = [
            {
                **i.to_dict(),
                'show_interact':False,
                'meta': (
                    user_info_public.from_orm(
                        db.query(user).filter(
                            or_(
                                and_(user.id != user_id, user.id == i.another),
                                and_(user.id != user_id, user.id == i.one)
                            )
                        ).first()
                    ) if i.room_type == 'double'
                    else Group_Room_Public.from_orm(db.query(Group_Room).filter_by(room_id=i.room_id).first())
                )
            }
            for i in db.query(Request_Room).filter(
                and_(
                    Request_Room.user_id != user_id,
                    or_(Request_Room.one == user_id, Request_Room.another == user_id),
                    Request_Room.status != 2,
                    Request_Room.status !=1)
            )
            .order_by(Request_Room.created_at.desc())
            .offset((limit - 1) * GET_ROOM_REQUEST_LENGTH)
            .limit(GET_ROOM_REQUEST_LENGTH)
            .all()
        ]

        return  {
            "data":Data,
            "meta":{
                "total":total_count,
                "size": GET_ROOM_REQUEST_LENGTH,
                'page':limit,
            }
        }
    except Exception as e:
        print(e)
        pass
    finally:
        pass





@mysocket_chat.post("/ws/set/room/response_to_request")
async def deal_response_to_request(ids:Union[list,int]=Body(...,embed=True),
                             deal:int=Body(...,embed=True),
                             user_id:int=Body(...,embed=True),
                             db:Session=Depends(get_db)):
    try:
        # 2过期,1处理成功,0出错
        room = (
            db.query(Request_Room).filter(
                or_(
                    and_(Request_Room.one == ids[0], Request_Room.another == ids[1]),
                    and_(Request_Room.one == ids[1], Request_Room.another == ids[0])
                )
            ).limit(1).first()
            if isinstance(ids, list)
            else db.query(Request_Room).filter_by(room_id=ids).first()
        )
        if room.status == 2:
            # 过期了
            return {'status': 200, 'result': 2}

        result = 1 if await deal_response_request_room(room.room_id, deal, user_id) else 0
        if result==1:
           await sender_message_user_online_room_request(user_id,room.room_id,room.room_type)
           if isinstance(ids, list):
               friend_id = next((i for i in ids if i != user_id), None)
               if friend_id is not None:
                   friend=db.query(user).filter_by(id=friend_id).first()
                   await sender_message_user_online_room_request(friend_id, room.room_id, room.room_type)
                   db.add(no_response(id=str(uuid.uuid4()),user_id=friend_id,title="好友通知",
                                      content=f"您申请的: 用户 {friend.username or friend.id } 通过了您的好友申请!",sender_info="系统通知"))
                   db.commit()
                   asyncio.create_task(sender_message_online_By_Type(friend_id,"render_notice"))
        return {'status': 200, 'result': result}

    except Exception as e:

        db.rollback()
        pass
    finally:
        db.close()




@mysocket_chat.post("/ws/set/room/add_user")
def deal_add_userinfo_to_room(room_id:str=Body(...),user_id:List=Body(...),type:str=Body(...)):
    """
    根据房间id和传输的用户信息批量处理,仅仅考虑group/middle，注意这个和单个群组申请区分
    """
    try:
        result=ADD_ROOM_USER_MAPPER(type)(user_id,room_id)
        if not result:
            return status.HTTP_404_NOT_FOUND
        return status.HTTP_200_OK
    except Exception as e:
        print(e)
        pass


@mysocket_chat.get("/ws/get/group_list")
def deal_group_get_user_list(user_id: int, db: Session = Depends(get_db)):
    """
    获取用户的群聊信息
    :param user_id:
    :param db:
    :return:
    """
    try:
        userexpand = db.query(user_expand).filter_by(id=user_id).first()
        if not userexpand:
            return status.HTTP_404_NOT_FOUND
        current_info = json.loads(userexpand.room_info or json.dumps({}))

        # 使用filter获取符合条件的房间ID列表
        room_ids = [r for r, _ in islice(current_info.items(), GROUP_USER_INFO_LENGTH)]
        # 批量查询群组信息
        groups = db.query(Group_Room).filter(Group_Room.room_id.in_(room_ids)).all()

        # 转换为Group_Room_Public对象并保持原始顺序
        DATA = []
        for r, _ in islice(current_info.items(), GROUP_USER_INFO_LENGTH):
            group = next((g for g in groups if g.room_id == r), None)
            if group:
                # 群组只返回房间信息更多信息另外获取
                DATA.append({
                    'room_info':  Group_Room_Public(**group.to_dict()),
                }
              )
        return DATA
    except Exception as e:
        print(e)
        raise  # 或者返回适当的错误响应
    finally:
        db.close()

@mysocket_chat.get("/ws/get/group/user_info")
def deal_get_group_user_info(room_id:str,limit:int=1,db:Session=Depends(get_db)):
    try:
        userexpand=db.query(Group_Room).filter_by(room_id=room_id).first()
        if not userexpand:
            return  status.HTTP_404_NOT_FOUND
        DATA=[]
        preuserids=list(json.loads(userexpand.user_info or json.dumps({})).items())
        userids=[item[0] for item in preuserids[(limit-1*GROUP_USER_INFO_LENGTH):limit*GROUP_USER_INFO_LENGTH]]
        for i in db.query(user).filter(user.id.in_(userids)).all():
            DATA.append(user_info_public.from_orm(i))
        return DATA

    except Exception as e:
        print(e)
        pass
    finally:
        db.close()



@mysocket_chat.get("/ws/get/friends_list")
def deal_get_user_list(user_id:int,db:Session=Depends(get_db)):
    """
    用于获取朋友列表,选择直接返回完毕，直接返回对应的房间信息,用于socket直接进行通讯
    :param user_id:
    :param db:
    :return:
    """
    try:
        userexpand=db.query(user_expand).filter_by(id=user_id).first()
        friend_info=[]
        if not userexpand:
            return status.HTTP_404_NOT_FOUND
        current_info=json.loads(userexpand.friends_info or json.dumps({}))

        ids=[]
        for r,c in current_info.items():
            # r好友id,在进行返回时,返回一下room_id，减少再用户不在线的redis时再次请求
            # 由于采用多张表,但没有直接关系
            ids.append(r)
        # 处理用户更新信息
        userinfo=db.query(user).filter(user.id.in_(ids)).all()
        user_info_dict={i.id :user_info_public.from_orm(i)  for i in userinfo}
        room_infos = (
            db.query(Double_Room)
            .filter(
                or_(
                    and_(Double_Room.one.in_(ids), Double_Room.another == user_id),
                    and_(Double_Room.one == user_id, Double_Room.another.in_(ids))
                )
            )
            .all()
        )
        for i in ids:
            # 注意这里i被转至str了
            userinfo = user_info_dict.get(int(i))
            if not userinfo:
                continue  # 如果用户信息不存在，跳过
            #过滤处相关的信息
            related_rooms = [
                room for room in room_infos
                if (room.one==int(i) and room.another==user_id) or (room.one==user_id and room.another== int(i))
            ]
            room_info_public_data = [
                Double_Room_Public.from_orm(room) for room in related_rooms
            ] if related_rooms else None
            # 在不出问题下理论是一条房间信息[]
            friend_info.append(
                {
                    "friend_info": user_info_public.from_orm(userinfo),
                    "room_info": room_info_public_data[0] if room_info_public_data else {},
                }
            )
        return friend_info
    except Exception as e:
        print(e)
        return status.HTTP_400_BAD_REQUEST

    finally:
        db.close()



@mysocket_chat.get("/ws/get/message/double/by_single")
def deal_get_group_message(room_id:str,
                           user_id:int,type:int=1,
                           room_stream:str=0,
                           limit:int=1,db:Session=Depends(get_db),redis: Redis = Depends(get_redis_client)):
    try:
        MESSAGE = []
        new_last_stream=str(0)
        if type:
            # 用于用户主动刷新消息
            # 获取全部的条目
            db.query(Double_Message).filter_by(room_id=room_id).offset((limit-1)*GET_MESSAGE_LENGTH).limit(GET_MESSAGE_LENGTH*limit).all()

        else:
            #获取更新流,消息
            userexpand = db.query(user_expand).filter_by(id=user_id).first()
            if not userexpand:
                return status.HTTP_404_NOT_FOUND
                # 每次读取时阻塞2s,等待前端2s同步,减少请求
            Ustream_info = json.loads(userexpand.last_stream or json.dumps({}))

            if room_stream:
                 last_stream =room_stream
            else:
                 last_stream=Ustream_info.setdefault(room_id, b'0')

            messages = redis.xread({room_id: last_stream}, block=1000)

            parse_data = parse_redis_data(messages)
            if parse_data:
                for i in parse_data.get("data"):
                    middledata=i.get('data')
                    middledata.setdefault('room_id',room_id)
                    # 由于装载问题此处添加
                    MESSAGE.append(middledata)
            new_last_stream = get_last_message_id(messages)
            pass
        db.commit()
        return {
            'data':MESSAGE,
            'meta':{
                'room_id':room_id,
                'stream_time':new_last_stream,
            }
        }
    except Exception as e:
        print(e)
    finally:
        db.close()
@mysocket_chat.post("/ws/get/message/double/all")
def deal_get_group_message(  user_id:int,room_info:dict=Body(...),
                            db: Session = Depends(get_db),
                            redis: Redis = Depends(get_redis_client)):
    """
    批量获取私聊的信息,但是流采用服务器的数据,不在单独验证。只做对当前聊天窗口的验证。
    :param room_id: 
    :param user_id: 
    :param db: 
    :param redis: 
    :return: 
    """
    try:
        MESSAGE = []
        for room_id,room, in room_info.items():
            userexpand = db.query(user_expand).filter_by(id=user_id).first()
            if not userexpand:
                MESSAGE.append({
                    'data': [],
                    'limit':1,
                    'meta': {
                        'room_id': room_id,
                        'stream_time': '0',
                        'type': "double"
                        # 新增字段type减少前端额外处理
                    }
                })
                continue
            Ustream_info = json.loads(userexpand.last_stream or '{}')
            if room.get('stream_time',None):
                last_stream = room['stream_time']
            else:
                last_stream = Ustream_info.get(room_id, '0')
            messages = redis.xread({room_id: last_stream}, block=2000)
            if not messages:
                MESSAGE.append({
                    'data': [],
                    'limit': 1,
                    'meta': {
                        'room_id': room_id,
                        'stream_time': last_stream,
                        'type':"double"
                         # 新增字段type减少前端额外处理
                    }
                })
                continue
            parse_data = parse_redis_data(messages)
            middle_message = [msg.get('data') for msg in parse_data.get("data", [])]
            new_last_stream = get_last_message_id(messages)
            MESSAGE.append({
                'data': middle_message,
                'limit':0,
                'meta': {
                    'room_id': room_id,
                    'stream_time': new_last_stream,
                    'type':"double"
                     # 新增字段type减少前端额外处理
                }
            })
            Ustream_info[room_id] = new_last_stream
            userexpand.last_stream = json.dumps(Ustream_info)

        db.commit()
        return MESSAGE

    except Exception as e:
        print(f"Error: {e}")
    finally:
        db.close()

@mysocket_chat.get("/ws/get/message/group/by_single")
def deal_get_group_messages(room_id:str,user_id:int,db:Session=Depends(get_db),redis: Redis = Depends(get_redis_client)):
    """
    后续需要拓展,分开写
    :param room_id:
    :param user_id:
    :param db:
    :param redis:
    :return:
    """
    try:
        userexpand=db.query(user_expand).filter_by(id=user_id).first()
        if not userexpand:
            return status.HTTP_404_NOT_FOUND
        Ustream_info = json.loads(userexpand.last_stream or json.dumps({}))
        #每次读取时阻塞2s,等待前端2s同步,减少请求
        last_stream =Ustream_info.setdefault(room_id, b'0')
        messages = redis.xread({room_id:  last_stream}, block=2000)
        MESSAGE=[]

        parse_data = parse_redis_data(messages)
        new_last_stream=last_stream
        if parse_data:
                for i in parse_data.get("data"):
                    middledata = i.get('data')
                    middledata.setdefault('room_id', room_id)
                    # 由于装载问题此处添加
                    MESSAGE.append(middledata)
                new_last_stream = get_last_message_id(messages)
        db.commit()
        return {
                'data': MESSAGE,
                'meta': {
                    'room_id': room_id,
                    'stream_time': new_last_stream,
                }
            }
    except Exception as e:
        db.rollback()
        pass
    finally:
        db.close()

@mysocket_chat.post("/ws/get/message/group/all")
def deal_get_group_messages(user_id:int,db:Session=Depends(get_db),room_info:dict=Body(...),redis: Redis = Depends(get_redis_client)):
    """
    后续需要拓展,分开写
    :param room_id:
    :param user_id:
    :param db:
    :param redis:
    :return:
    """
    try:
        userexpand=db.query(user_expand).filter_by(id=user_id).first()
        if not userexpand:
            return status.HTTP_404_NOT_FOUND
        Ustream_info = json.loads(userexpand.last_stream or json.dumps({}))
        #每次读取时阻塞2s,等待前端2s同步,减少请求
        MESSAGE=[]

        for room_id,room in room_info.items():

            if  room.get('stream_time',None):
                last_stream = room['stream_time']

            else:
                last_stream = Ustream_info.setdefault(room_id, b'0')
            messages = redis.xread({room_id: last_stream}, block=2000)
            parse_data = parse_redis_data(messages)
            new_last_stream = last_stream

            middle_message=[]

            if parse_data:
                for i in parse_data.get("data"):
                    middle_message.append(i.get('data'))
                new_last_stream = get_last_message_id(messages)

            MESSAGE.append({
                        'data': middle_message,
                        'limit': 0 if len(middle_message) > 0 else 1,
                        'meta': {
                            'room_id': room_id,
                            'stream_time': new_last_stream,
                            'type': 'group'
                            # 新增字段type为了减少前端的额外处理
                    }
            })

        db.commit()
        return MESSAGE
    except Exception as e:
        print(1)
        print(e)
        db.rollback()
        return status.HTTP_500_INTERNAL_SERVER_ERROR
        pass
    finally:
        db.close()


@mysocket_chat.post("/ws/flush/room_stream")
def deal_change_message_stream_meta(user_id:int,stream_time:str,
                                    room_id:Union[str,dict]=Body(...,embed=True),
                                    type:int=1,db:Session=Depends(get_db)):
    """
    确认更新stream流
    :param room_id:
    :param stream_time:
    :param user_id:
    :param db:
    :return:
    """
    try:
        userexpand = db.query(user_expand).filter_by(id=user_id).first()
        if not userexpand:
                return status.HTTP_404_NOT_FOUND
        Ustream_info = json.loads(userexpand.last_stream or json.dumps({}))
        if type:
            Ustream_info[room_id]=stream_time
            userexpand.last_stream = json.dumps(dict(Ustream_info))
            db.commit()
        else:
            for i in room_id:
                Ustream_info[i.room_id]=Ustream_info[i.stream_time]
            userexpand.last_stream = json.dumps(Ustream_info)
            #批量处理
            db.commit()
        return status.HTTP_200_OK

    except  Exception as e:
        print(e)
        return status.HTTP_500_INTERNAL_SERVER_ERROR
    finally:
        db.close()


@mysocket_chat.post("/ws/user/search/group_user")
async def deal_search_user_friend(
        user_id: int,
        key: Union[str, int],
        limit: int = 1,
        db: Session = Depends(get_db)
):
    """
    搜索用户/群组，注意用户返回的是用户信息,群组才是房间的信息
    :param user_id: 当前用户ID
    :param key: 搜索关键词（用户名或ID）
    :param limit: 页码（从1开始）
    :param db: 数据库会话
    :return: 搜索结果和分页信息
    """
    try:
        offset = (limit - 1) * GET_USER_SEARCH_LENGTH

        # 构建用户的查询条件
        user_query = db.query(user).filter(
            user.id != user_id  # 排除自己
        )

        # 如果key是数字，优先匹配ID
        if isinstance(key, int) or (isinstance(key, str) and key.isdigit()):
            user_query = user_query.filter(
                (user.id == int(key)) |
                (user.username.ilike(f"%{key}%"))
            )
        else:
            # 只匹配用户名
            user_query = user_query.filter(user.username.ilike(f"%{key}%"))
        #获取用户的好友群组信息用于,区分效果是否是好友
        userexpand=db.query(user_expand).filter_by(id=user_id).first()
        friend_ids=[int(i) for i in json.loads(userexpand.friends_info or json.dumps({})).keys()]
        group_ids=json.loads(userexpand.room_info or json.dumps({})).keys()

        # 获取用户查询结果

        user_results = user_query.offset(offset).limit(GET_USER_SEARCH_LENGTH).all()
        user_count = user_query.count()
        # 构建群组的查询条件
        group_query = db.query(Group_Room).filter(
            (Group_Room.room_id == int(key)) |
            (Group_Room.group_name.ilike(f"%{key}%")) if isinstance(key, int) or (isinstance(key, str) and key.isdigit())
            else Group_Room.group_name.ilike(f"%{key}%")
        )

        # 获取群组查询结果
        group_results = group_query.offset(offset).limit(GET_USER_SEARCH_LENGTH).all()
        group_count = group_query.count()

        # 合并结果并标记来源
        # show_info字段用于前端交互效果
        combined_results = [
            {"type": "user",'id':u.id,'collected': (u.id in friend_ids) , 'show_info':False, "data": user_info_public.from_orm(u)} for u in user_results
        ] + [
            {"type": "group", 'id':g.room_id,'collected': (g.room_id in group_ids), 'show_info':False,"data": Group_Room_Public(**(g.to_dict()))} for g in group_results
        ]
        # 合并总数
        total_count = user_count + group_count
        return {
            "data": combined_results,
            # meta,减少前端计算频率
            "pagination": {
                'total': total_count,
                "limit": limit,
                "page_size": GET_USER_SEARCH_LENGTH,
                "total_pages": (total_count + GET_USER_SEARCH_LENGTH - 1) // GET_USER_SEARCH_LENGTH
            }
        }
    except Exception as e:
        print(e)
        pass
    finally:
        db.close()



@mysocket_chat.post("/ws/user/user_group/search")
def search_user_or_group(
        user_id: int,
        key: Union[str, int],
        limit: int = 1,
        db: Session = Depends(get_db)
):
    """
    搜索好友的用户或群组,注意用户返回的是用户信息,群组才是房间的信息
    :param user_id: 当前用户ID
    :param key: 搜索关键词（用户名或ID）
    :param limit: 页码（从1开始）
    :param db: 数据库会话
    :return: 搜索结果和分页信息
    """
    try:

        userexpand = db.query(user_expand).filter_by(id=user_id).first()
        if not userexpand:
            return {
            "data":[],
            "pagination": {
                "total":0,
                "limit": limit,
                "page_size": GET_USER_SEARCH_LENGTH,
                "total_pages": 0
            }
        }

        room_info = json.loads(userexpand.room_info or "{}")
        friends_info = json.loads(userexpand.friends_info or "{}")


        group_ids = [id for id, meta in room_info.items() if meta.get('type') == 'group']
        friend_ids =[ int(i) for i in friends_info.keys()]

        # 查询好友信息
        if key.isdigit() and int(key) in friend_ids:
            user_query = db.query(user).filter(
                user.id != user_id,
                (user.id == int(key)) | (user.username.ilike(f"%{key}%"))
            )
        else:
            user_query = db.query(user).filter(
                user.id != user_id,
                user.id.in_(friend_ids),
                user.username.ilike(f"%{key}%")
            )
        user_results = user_query.offset((limit - 1) * GET_USER_SEARCH_LENGTH).limit(GET_USER_SEARCH_LENGTH).all()
        user_count = user_query.count()

        # 群组的搜索
        if key.isdigit() and int(key) in group_ids:
            group_query = db.query(Group_Room).filter(
                (Group_Room.room_id == int(key)) |
                (Group_Room.group_name.ilike(f"%{key}%"))
            )
        else:

            group_query = db.query(Group_Room).filter(
                Group_Room.group_name.ilike(f"%{key}%")
            )

        group_results = group_query.offset((limit - 1) * GET_USER_SEARCH_LENGTH).limit(GET_USER_SEARCH_LENGTH).all()
        group_count = group_query.count()


        combined_results = [
            {"type": "user",'collected':True, 'id':i.id,'show_info':False, "data": user_info_public.from_orm(i)} for i in user_results
        ] + [
            #这里由于架构问题,处理一下
            {"type": "group", 'collected': True,'id':group.room_id,'show_info':False, "data": Group_Room_Public(**group.to_dict())} for group in group_results
            if bool(json.loads(group.user_info or json.dumps({})).get(user_id))
        ]

        response_data = {
            "data": combined_results,
            "pagination": {
                "total": user_count + group_count,
                "limit": limit,
                "page_size": GET_USER_SEARCH_LENGTH,
                "total_pages": (user_count + group_count + GET_USER_SEARCH_LENGTH - 1) // GET_USER_SEARCH_LENGTH
            }
        }
        return response_data

    except Exception as e:
        print(e)
        return status.HTTP_400_BAD_REQUEST
    finally:
        db.close()





@mysocket_chat.post("/ws/room/get_info/by/room_id_type")
def deal_get_room_info_by_id_type(user_id:int,type:str,room_ids:list=Body(...),db:Session=Depends(get_db)):
    """
    用于更新历史对话的房间信息,实际是通过好友列表的信息获取,预留可拓展type
    :param room_id:
    :param type:
    :param db:
    :return:
    """
    try:
        if type=="group":
            RoomList=db.query(Group_Room).filter(Group_Room.room_id.in_(room_ids)).all()
            DATA={}
            for i in RoomList:
                DATA.setdefault(i.room_id,{'room_info': Group_Room_Public(**i.to_dict())})
            return DATA
        if type=="double":
            RoomList = db.query(Double_Room).filter(Double_Room.room_id.in_(room_ids)).all()
            userids=[ i.one if i.another==user_id else i.another for i in RoomList]
            users={ i.id : i  for i in  db.query(user).filter(user.id.in_(userids)).all()}
            DATA = {
                i.room_id:{
                'friend_info': user_info_public.from_orm(users[i.one if i.one != user_id else i.another]),
                'room_info': Double_Room_Public.from_orm(i)
            } for i in RoomList
            }
            return  DATA
    except Exception as e:
        print(e)
        return status.HTTP_404_NOT_FOUND
    finally:
        db.close()
@mysocket_chat.get("/ws/room/get_ids/all")
def deal_get_room_ids_by_user_id(user_id:int,db:Session=Depends(get_db)):
    try:
        userexpand=db.query(user_expand).filter_by(user_id=user_id).first()
        data=[i for i in json.loads(userexpand.room_info or json.dumps({}))]
        return data
    except Exception as e:
        pass
    finally:
        db.close()



@mysocket_chat.post("/ws/room/get_info/all/by/room_id_type")
def deal_get_room_info_by_id_type(type:str,room_id:str,user_id:int=None,db:Session=Depends(get_db)):
    """
    用于单个获取服务器列表的房间信息
    :param room_id:
    :param type:
    :param db:
    :return:
    """
    try:
        DATA=None
        if type=="group":
            DATA=db.query(Group_Room).filter_by(room_id=room_id).first()
            if DATA:
                DATA={
                    "room_info":Group_Room_Public(**DATA.to_dict())
                }

        if type=="double":
            DATA = db.query(Double_Room).filter_by(room_id=room_id).first()
            friend_info=db.query(user).filter(and_(
                    user.id.in_([DATA.one ,DATA.another]
                ),user.id!=user_id)).first()
            if DATA:
                DATA= {
                    "friend_info": user_info_public.from_orm(friend_info),
                    "room_info": Double_Room_Public.from_orm(DATA),
                }
        return  DATA
    except Exception as e:
        print(e)
        return status.HTTP_404_NOT_FOUND
    finally:
        db.close()

@mysocket_chat.get("/ws/get/about/message/noresponse")
async def deal_get_about_message(user_id:int,limit:int=1,db:Session=Depends(get_db)):
    try:
        total_count=db.query(no_response).filter_by(user_id=user_id).count()
        data=(db.query(no_response).filter_by(user_id=user_id)
            .offset((limit-1)*GET_MESSAGE_NORESPONSE_LENGTH)
            .limit(GET_MESSAGE_NORESPONSE_LENGTH).all())
        return {
                "data":data,
                'meta': {
                    'total':total_count,
                    'current':limit,
                }
             }
    except Exception as e:
        pass
    finally:
        db.close()

@mysocket_chat.post("/ws/delete/about/message/noresponse")
def deal_delete_no_response_message(user_id:int,message_id:str,db:Session=Depends(get_db)):
    try:
        db.query(no_response).filter_by(user_id=user_id,id=message_id).delete()
        db.commit()
        return status.HTTP_200_OK
    except Exception as e:
        return  status.HTTP_404_NOT_FOUND
    finally:
        db.close()







